package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwdTradeSource {
    public static void main(String[] args) {

        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //1.3 指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 2.检查点相关的设置(略)
        //TODO 3.从kafka的topic_db主题中读取数据 创建动态表
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_trade_order_source"));
        //TODO 4.过滤出支付表数据
        Table paymentInfo = tableEnv.sqlQuery("select \n" +
                "data['id'] id,\n" +
                "data['order_id'] order_id,\n" +
                "data['total_amount'] total_amount,\n" +
                "data['payment_type'] payment_type,\n" +
                "date_format(data['create_time'],'yyyy-MM-dd') date_id,\n" +
                "data['create_time'] create_time\n" +
                "from `topic_db` where `table` = 'payment_info' " +
                "and `type` = 'insert'\n");
        tableEnv.createTemporaryView("payment_info", paymentInfo);

        //TODO 5.过滤出订单表数据
        Table orderInfo = tableEnv.sqlQuery("select \n" +
                "data['id'] id,\n" +
                "data['user_id'] user_id\n" +
                "from `topic_db` where `table` = 'order_info' " +
                "and `type` = 'insert'\n");
        tableEnv.createTemporaryView("order_info", orderInfo);

        //TODO 7.将上述2张表进行关联
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "oi.id order_id,\n" +
                "oi.user_id user_id,\n" +

                "pi.id payment_id,\n" +
                "pi.total_amount total_amount,\n" +
                "pi.payment_type payment_type,\n" +
                "pi.date_id date_id,\n" +
                "pi.create_time create_time\n" +

                "from order_info oi \n" +
                "join payment_info pi\n" +
                "on oi.id = pi.order_id\n"

                );
        tableEnv.createTemporaryView("result_table", resultTable);

        //TODO 8.创建动态表 和要写入的kafka主题进行映射

        tableEnv.executeSql("create table dwd_trade_order_source(\n" +
                "order_id string,\n" +
                "user_id string,\n" +
                "payment_id string,\n" +
                "total_amount string,\n" +
                "payment_type string,\n" +
                "date_id string,\n" +
                "create_time string,\n" +

                "primary key(order_id) not enforced\n" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_order_source"));

        //TODO 9.写入
        tableEnv.executeSql("insert into dwd_trade_order_source select * from result_table");

    }
}
